package blue.mqtt4j.server.net;

import blue.mqtt4j.core.util.MessageIdGen;
import blue.mqtt4j.server.topic.TopicChannel;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageBuilders;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttPublishVariableHeader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;

/**
 * @author Jin Zheng
 * @since 2020-03-22
 */
public class MqttPublishHandler extends SimpleChannelInboundHandler<MqttPublishMessage>
{
	private static Logger logger = LoggerFactory.getLogger(MqttPublishHandler.class);

	private final TopicChannel topicChannel;

	public MqttPublishHandler(TopicChannel topicChannel)
	{
		this.topicChannel = topicChannel;
	}

	@Override
	protected void channelRead0(ChannelHandlerContext ctx, MqttPublishMessage message) throws Exception
	{
		MqttFixedHeader reqHeader = message.fixedHeader();
		MqttPublishVariableHeader variableHeader = message.variableHeader();
		ByteBuf payload = message.payload();
		logger.info("header: {}", reqHeader);
		logger.info("variableHeader: {}", variableHeader);
		logger.info("payload: {}", payload.toString(StandardCharsets.UTF_8));

		MqttMessage ackMessage = MqttMessageBuilders.pubAck()
				.packetId((short)variableHeader.packetId())
				.reasonCode((byte)0)
				.build();
		ctx.writeAndFlush(ackMessage);
		System.out.println(message.refCnt());

		MqttPublishMessage publishMessage = MqttMessageBuilders.publish()
				.messageId(MessageIdGen.getInstance().id())
				.payload(payload)
				.topicName(variableHeader.topicName())
				.qos(reqHeader.qosLevel())
				.build();
		topicChannel.write(variableHeader.topicName(), publishMessage);
	}
}
